package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;

public class Producer {
    public static void main(String[] args) throws MQClientException {
        //创建"事务消息生产者"
        TransactionMQProducer p = new TransactionMQProducer("producer-group-3");

        //设置nameserver
        p.setNamesrvAddr("192.168.64.141:9876");

        //设置监听器,只要已发送事务消息,就会触发监听器执行本地事务,
        // 另外还要处理服务器的事务回查
        p.setTransactionListener(new TransactionListener() {
            /*
            事务id ---- 事务的执行状态
             */
            ConcurrentHashMap<String , LocalTransactionState> map=
                    new ConcurrentHashMap<>();

            //执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
                System.out.println("执行本地事务");

                //模拟网络中断,无法通知rocketmq事务执行状态
                if (Math.random()<1){
                    System.out.println("由于网络故障无法通知rocketmq");
                    map.put(msg.getTransactionId(), LocalTransactionState.UNKNOW);
                    return LocalTransactionState.UNKNOW;
                }

                //模拟成功或失败
                if (Math.random()<0.5){
                    System.out.println("成功");
                    map.put(msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE);
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else {
                    System.out.println("失败");
                    map.put(msg.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
            //处理事务状态回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("服务器正在回查事务状态");
                return map.get(msg.getTransactionId());
            }
        });

        //启动
        p.start();

        //发送事务消息
        while (true){
            System.out.println("输入消息:");
            String s=new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            // 只需要一发送事务消息,就会触发监听器执行本地事务,即业务操作
            // 第二个参数会传递到监听器执行业务运算
            // p.sendMessageInTransaction(msg, 传递给监听器的业务数据对象);
            //半消息
            TransactionSendResult r = p.sendMessageInTransaction(msg, null);
            System.out.println("事务消息处理完成:"+r);
        }
    }
}
